﻿using System.Data;
using Dapper;
using RockBus.DataContracts;
using RockBus.Logging.Interfaces;
using RockBus.MessageRepository.Interfaces;
using System;
using System.Collections.Generic;
using System.Data.SqlClient;
using System.Linq;
using System.Reflection;
using System.Text;
using System.Threading.Tasks;
using System.Transactions;

namespace RockBus.MessageRepository.Dapper.Db
{
    public class MessageRepository : IMessageRepository
    {
        public MessageRepository(ILogger logger, IDbMessageRepositoryConfiguration configuration)
        {
            this.Logger = logger;
            this.Configuration = configuration;
        }

        public ILogger Logger { get; set; }

        public IDbMessageRepositoryConfiguration Configuration { get; set; }

        public long CreateMessage(
            string correlationId,
            string messageVersion,
            string action,
            DateTime receiveTime,
            string receiveAddress)
        {
            IDbConnection dbConn = new SqlConnection(this.Configuration.ConnectionString);

            try
            {
                dbConn.Open();

                long messageId = 0;
                string sqlCmdIns = "insert Message (CorrelationId " +
                                                    ",MessageVersion " +
                                                    ",Action " +
                                                    ",ReceiveTime " +
                                                    ",ReceiveAddress " +
                                                    ",CreateTime) " +
                                    "values (@correlationId " +
                                            ",@messageVersion " +
                                            ",@action " +
                                            ",@receiveTime " +
                                            ",@receiveAddress " +
                                            ",sysutcdatetime()); " +
                                    "select cast(scope_identity() as bigint)";
                messageId = dbConn.Query<long>(sqlCmdIns, new
                {
                    correlationId = correlationId,
                    messageVersion = messageVersion,
                    action = action,
                    receiveTime = receiveTime,
                    receiveAddress = receiveAddress
                }).Single();

                return messageId;
            }
            catch(Exception exc)
            {
                this.Logger.Log(MethodBase.GetCurrentMethod(), exc);
                throw;
            }
            finally
            {
                if(dbConn != null)
                    dbConn.Dispose();
            }
        }

        public long CreateMessageFragment(
            long messageId,
            int partIndex,
            byte[] body,
            int size)
        {
            IDbConnection dbConn = new SqlConnection(this.Configuration.ConnectionString);

            try
            {
                dbConn.Open();

                string sqlCmdIns = "insert MessageFragment (MessageId " +
                                                          ",PartIndex " +
                                                          ",Body " +
                                                          ",Size) " +
                                   "values (@messageId " +
                                          ",@partIndex " +
                                          ",@bodyPrmt " +
                                          ",@sizePrmt); " +
                                   "select cast(scope_identity() as bigint)";

                byte[] output = new byte[size];
                Buffer.BlockCopy(body, 0, output, 0, size);

                return dbConn.Query<long>(sqlCmdIns, new
                {
                    messageId = messageId,
                    partIndex = partIndex,
                    bodyPrmt = output,
                    sizePrmt = size
                }).Single();
            }
            catch(Exception exc)
            {
                this.Logger.Log(MethodBase.GetCurrentMethod(), exc);
                throw;
            }
            finally
            {
                if(dbConn != null)
                    dbConn.Dispose();
            }
        }

        public long EnqueueEntry(
            long messageId,
            byte[] routingHeaderByteArray,
            int priority,
            bool orderedDelivery,
            string receiverMachineName,
            int receiverProcessId,
            int receiverAppDomainId,
            long? validityDurationSecs,
            long deliveryAttemptIntervalSecs,
            int maxDeliveryAttemptCount,
            bool isResponse)
        {
            IDbConnection dbConn = new SqlConnection(this.Configuration.ConnectionString);

            try
            {
                dbConn.Open();

                string sqlCmdIns = "insert QueueEntry (MessageId " +
                                                     ",RoutingHeader " +
                                                     ",Priority " +
                                                     ",OrderedDelivery " +
                                                     ",ReceiverMachineName " +
                                                     ",ReceiverProcessId " +
                                                     ",ReceiverAppDomainId " +
                                                     ",EnqueueTime " +
                                                     ",ExpirationTime " +
                                                     ",StateId " +
                                                     ",DequeueAttemptCount " +
                                                     ",DeliveryAttemptCount " +
                                                     ",DeliveryAttemptIntervalSecs " +
                                                     ",MaxDeliveryAttemptCount " +
                                                     ",IsResponse) " +
                                   "values (@messageId " +
                                            ",@routingHeader " +
                                            ",@priority " +
                                            ",@orderedDelivery " +
                                            ",@receiverMachineName " +
                                            ",@receiverProcessId " +
                                            ",@receiverAppDomainId " +
                                            ",sysutcdatetime() " +
                                            ",dateadd(second, cast(@messageValidityDurationSecs as int), sysutcdatetime()) " +
                                            ",0 " + // StateId
                                            ",0 " + // DequeueAttemptCount
                                            ",0 " + // DeliveryAttemptCount
                                            ",@deliveryAttemptIntervalSecs " +
                                            ",@maxDeliveryAttemptCount " +
                                            ",@isResponse); " +
                                    "select cast(scope_identity() as bigint)";

                long queueEntryId = 0;

                object messageVDS = DBNull.Value;
                if(validityDurationSecs.HasValue)
                    messageVDS = validityDurationSecs.Value;

                queueEntryId = dbConn.Query<long>(sqlCmdIns, new
                {
                    messageId = messageId,
                    routingHeader = routingHeaderByteArray,
                    priority = priority,
                    orderedDelivery = orderedDelivery,
                    receiverMachineName = receiverMachineName,
                    receiverProcessId = receiverProcessId,
                    receiverAppDomainId = receiverAppDomainId,
                    messageValidityDurationSecs = messageVDS,
                    deliveryAttemptIntervalSecs = deliveryAttemptIntervalSecs,
                    maxDeliveryAttemptCount = maxDeliveryAttemptCount,
                    isResponse = isResponse
                }).Single();

                return queueEntryId;
            }
            catch(Exception exc)
            {
                this.Logger.Log(MethodBase.GetCurrentMethod(), exc);
                throw;
            }
            finally
            {
                if(dbConn != null)
                    dbConn.Dispose();
            }
        }

        public void CompleteMessage(
            long messageId,
            int partCount,
            int size)
        {
            IDbConnection dbConn = new SqlConnection(this.Configuration.ConnectionString);

            try
            {
                dbConn.Open();

                Logger.Log(MethodBase.GetCurrentMethod(), LogLevelEnum.Debug, Transaction.Current);

                string sqlCmdUpd = "update Message with (updlock) " +
                                    "set PartCount = @partCount " +
                                    "   ,Size      = @sizePrmt " +
                                    "where Id = @messageId; " +

                                    "update QueueEntry with (updlock) " +
                                    "set StateId = 1 " +
                                    "where MessageId = @messageId " +
                                    "  and StateId   = 0 ";

                dbConn.Execute(sqlCmdUpd, new
                {
                    partCount = partCount,
                    sizePrmt = size,
                    messageId = messageId
                });

            }
            catch(Exception exc)
            {
                this.Logger.Log(MethodBase.GetCurrentMethod(), exc);
                throw;
            }
            finally
            {
                if(dbConn != null)
                    dbConn.Dispose();
            }
        }

        public byte[] GetMessageFragment(
            long messageId,
            int partIndex)
        {
            IDbConnection dbConn = new SqlConnection(this.Configuration.ConnectionString);

            try
            {
                dbConn.Open();

                string sqlCmdSel = "select m.Body " +
                                   "from MessageFragment m with (nolock) " +
                                   "where m.MessageId = @messageId " +
                                   "  and m.PartIndex = @partIndex; ";
                byte [] bArr = null;

                IEnumerable<byte[]> byteList = dbConn.Query<byte[]>(sqlCmdSel, new
                {
                    messageId = messageId,
                    partIndex = partIndex
                });

                foreach(byte [] byteArray in byteList)
                {
                    if(byteArray == null)
                    {
                        this.Logger.LogError(MethodBase.GetCurrentMethod(), "MessageFragment could not be read. MessageId: {0}; PartIndex: {1}.", messageId.ToString(), partIndex.ToString());
                        return null;
                    }

                    bArr = byteArray;
                }

                return bArr;
            }
            catch(Exception exc)
            {
                this.Logger.Log(MethodBase.GetCurrentMethod(), exc);
                throw;
            }
        }

        public Task<DequeueResult> StartDequeueEnties(
            int maxEntryCount,
            string machineName,
            int processId,
            int appDomainId,
            string processorId)
        {
            try
            {
                return Task.Factory.StartNew<DequeueResult>(() => Dequeue(
                    maxEntryCount,
                    machineName,
                    processId,
                    appDomainId,
                    processorId
                ));
            }
            catch(Exception exc)
            {
                this.Logger.Log(MethodBase.GetCurrentMethod(), exc);
                throw;
            }
        }

        public DequeueResult Dequeue(
            int maxEntryCount,
            string machineName,
            int processId,
            int appDomainId,
            string processorId)
        {
            Guid batchId = Guid.NewGuid();
            int returnCode = 0;

            List<QueueEntry> queueEntries = new List<QueueEntry>();

            SqlConnection dbConn = new SqlConnection(this.Configuration.ConnectionString);

            #region query
            string sqlCmd = "update top(@count) q with (updlock, rowlock, readpast) " +
                            "set q.StateId = 2 " +
                            "   ,q.MachineName =  @machineName " +
                            "   ,q.ProcessId = @processId " +
                            "   ,q.ProcessorId = @processorId " +
                            "   ,q.BatchId = @batchId " +
                            "   ,q.DequeueTime = sysutcdatetime() " +
                            "   ,q.DequeueAttemptCount = q.DequeueAttemptCount + 1 " +
                            "from QueueEntry q " +
                            "where q.StateId = 1 " +
                            "  and q.BatchId is null " +
                            "  and q.DeliveryAttemptCount < q.MaxDeliveryAttemptCount " +
                            " and (q.ExpirationTime is null " +
                            "   or q.ExpirationTime < sysutcdatetime()) " +
                            "and ((q.IsResponse = 0) " +
                            "  or (q.IsResponse = 1 " +
                            "  and q.ReceiverMachineName = @machineName " +
                            "  and q.ReceiverProcessId = @processId " +
                            "  and q.ReceiverAppDomainId = @appDomainId)); " +

                            "select q.Id " +
                            "      ,q.MessageId " +
                            "      ,q.ExpirationTime " +
                            "      ,q.StateId " +
                            "      ,q.RoutingHeader " +
                            "      ,q.Priority " +
                            "      ,q.OrderedDelivery " +
                            "      ,q.EnqueueTime " +
                            "      ,q.ReceiverMachineName " +
                            "      ,q.ReceiverProcessId " +
                            "      ,q.ReceiverAppDomainId " +
                            "      ,q.MachineName " +
                            "      ,q.ProcessId " +
                            "      ,q.ProcessorId " +
                            "      ,q.BatchId " +
                            "      ,q.DequeueTime " +
                            "      ,q.DequeueAttemptCount " +
                            "      ,q.DeliveryAttemptCount " +
                            "      ,q.IsResponse " +
                            "      ,m.PartCount " +
                            "      ,m.Size " +
                            "      ,m.CorrelationId " +
                            "      ,m.MessageVersion" +
                            "      ,m.Action " +
                            "      ,m.ReceiveTime " +
                            "      ,m.ReceiveAddress " +
                            "      ,m.CreateTime " +
                            "from QueueEntry q with (nolock) " +
                            "    ,Message m with (nolock) " +
                            "where q.BatchId = @batchId " +
                            "  and q.MessageId = m.Id " +
                            "order by q.Priority desc " +
                            "        ,q.Id asc; ";
            #endregion

            string transName = "dequeue";

            dbConn.Open();
            SqlTransaction dbTrans = dbConn.BeginTransaction(System.Data.IsolationLevel.ReadCommitted, transName);
            try
            {
                IEnumerable<dynamic> qeList = dbConn.Query(sqlCmd, new
                {
                    count = maxEntryCount,
                    machineName = machineName,
                    processId = processId,
                    appDomainId = appDomainId,
                    processorId = processorId,
                    batchId = batchId
                }, dbTrans);

                if(qeList.Count() == 0)
                {
                    this.Logger.LogDebug(MethodBase.GetCurrentMethod(), "No entries to dequeue.");
                    return null;
                }

                foreach(var qEntry in qeList)
                {
                    QueueEntry qe = new QueueEntry();

                    qe.Id = qEntry.Id;
                    qe.MessageId = qEntry.MessageId;
                    qe.ExpirationTime = qEntry.ExpirationTime;
                    qe.StateId = qEntry.StateId;
                    qe.RoutingHeader = RoutingHeader.CreateFromByteArray((byte[])qEntry.RoutingHeader);
                    qe.Priority = qEntry.Priority;
                    qe.OrderedDelivery = qEntry.OrderedDelivery;
                    qe.EnqueueTime = qEntry.EnqueueTime;
                    qe.ReceiverMachineName = qEntry.ReceiverMachineName;
                    qe.ReceiverProcessId = qEntry.ReceiverProcessId;
                    qe.ReceiverAppDomainId = qEntry.ReceiverAppDomainId;
                    qe.MachineName = qEntry.MachineName;
                    qe.ProcessId = qEntry.ProcessId;
                    qe.ProcessorId = qEntry.ProcessorId;
                    qe.BatchId = qEntry.BatchId;
                    qe.DequeueTime = qEntry.DequeueTime;
                    qe.DequeueAttemptCount = qEntry.DequeueAttemptCount;
                    qe.DeliveryAttemptCount = qEntry.DeliveryAttemptCount;
                    qe.IsResponse = qEntry.IsResponse;
                    qe.PartCount = qEntry.PartCount;
                    qe.Size = qEntry.Size;
                    qe.CorrelationId = qEntry.CorrelationId;
                    qe.MessageVersion = qEntry.MessageVersion;
                    qe.Action = qEntry.Action;
                    qe.ReceiveTime = qEntry.ReceiveTime;
                    qe.ReceiveAddress = qEntry.ReceiveAddress;
                    qe.CreateTime = qEntry.CreateTime;

                    queueEntries.Add(qe);
                }

                dbTrans.Commit();
            }
            catch(Exception ex)
            {
                dbTrans.Rollback(transName);
                batchId = Guid.Empty;
                returnCode = 1;
                this.Logger.LogError(MethodBase.GetCurrentMethod(), "Exception-msg: {0}", ex.Message);
            }
            finally
            {
                if(dbConn != null)
                    dbConn.Dispose();

                if(dbTrans != null)
                    dbTrans.Dispose();
            }

            if(returnCode != 0)
            {
                this.Logger.LogError(MethodBase.GetCurrentMethod(), "Return code was {0}.", returnCode);
                return null;
            }

            if(batchId == null || batchId == Guid.Empty)
            {
                this.Logger.LogError(MethodBase.GetCurrentMethod(), "BatchId was NULL.");
                return null;
            }

            DequeueResult result = new DequeueResult(batchId, queueEntries);
            return result;
        }

        public void CompleteEntry(
            long queueEntryId)
        {
            IDbConnection dbConn = new SqlConnection(this.Configuration.ConnectionString);

            try
            {
                dbConn.Open();

                Logger.Log(MethodBase.GetCurrentMethod(), LogLevelEnum.Debug, Transaction.Current);

                string sqlCmdUpd = "update q with (updlock) " +
                                    "set q.StateId = 3 " +
                                    "   ,q.DeliveryAttemptCount = DeliveryAttemptCount + 1 " +
                                    "   ,q.DeliveryTime = sysutcdatetime() " +
                                    "from QueueEntry q " +
                                    "where q.Id = @queueEntryId ";
                dbConn.Execute(sqlCmdUpd, new
                {
                    queueEntryId = queueEntryId
                });
            }
            catch(Exception exc)
            {
                this.Logger.Log(MethodBase.GetCurrentMethod(), exc);
                throw;
            }
        }

        public void RollbackEntry(
            long queueEntryId,
            Exception exception)
        {
            IDbConnection dbConn = new SqlConnection(this.Configuration.ConnectionString);

            try
            {
                dbConn.Open();

                string sqlCmdUpd = "update q with (rowlock) " +
                                    "set q.StateId = 1 " +
                                    "   ,q.MachineName = null " +
                                    "   ,q.ProcessId = null " +
                                    "   ,q.ProcessorId = null " +
                                    "   ,q.BatchId = null " +
                                    "   ,q.DeliveryAttemptCount = q.DeliveryAttemptCount + 1 " +
                                    "   ,q.DeliveryTime = null " +
                                    "   ,q.ErrorMessage = @errorMsg " +
                                    "from QueueEntry q " +
                                    "where q.Id = @queueEntryId ";

                dbConn.Execute(sqlCmdUpd, new
                {
                    errorMsg = exception.ToString(),
                    queueEntryId = queueEntryId
                });
            }
            catch(Exception exc)
            {
                this.Logger.Log(MethodBase.GetCurrentMethod(), exc);
                throw;
            }
        }

        public void FaultEntry(
            long queueEntryId,
            Exception exception)
        {
            IDbConnection dbConn = new SqlConnection(this.Configuration.ConnectionString);

            try
            {
                dbConn.Open();
                string sqlCmdUpd = "update q with (rowlock) " +
                                    "set q.StateId = 4 " +
                                    "   ,q.BatchId = null " +
                                    "   ,q.DeliveryAttemptCount = q.DeliveryAttemptCount + 1 " +
                                    "   ,q.DeliveryTime = null " +
                                    "   ,q.ErrorMessage = @errorMsg " +
                                    "from QueueEntry q " +
                                    "where q.Id = @queueEntryId ";

                dbConn.Execute(sqlCmdUpd, new
                {
                    errorMsg = exception.ToString(),
                    queueEntryId = queueEntryId
                });
            }
            catch(Exception exc)
            {
                this.Logger.Log(MethodBase.GetCurrentMethod(), exc);
                throw;
            }
        }

        public void CreateErrorMessage(
            long messageId,
            byte[] routingHeaderByteArray,
            string machineName,
            int processId,
            int appDomainId,
            int deliveryAttemptCount,
            Exception exception)
        {
            IDbConnection dbConn = new SqlConnection(this.Configuration.ConnectionString);
            try
            {
                dbConn.Open();

                string sqlCmdIns = "insert ErrorMessage (MessageId " +
                                                        ",RoutingHeader " +
                                                        ",MachineName " +
                                                        ",ProcessId " +
                                                        ",AppDomainId " +
                                                        ",DeliveryAttemptCount " +
                                                        ",ErrorMessage " +
                                                        ",CreateTime) " +
                                    "values (@messageId " +
                                            ",@routingHeader " +
                                            ",@machineName " +
                                            ",@processId " +
                                            ",@appDomainId " +
                                            ",@deliveryAttemptCount " +
                                            ",@errorMsg " +
                                            ",sysutcdatetime()) ";

                dbConn.Execute(sqlCmdIns, new
                {
                    messageId = messageId,
                    routingHeader = routingHeaderByteArray,
                    machineName = machineName,
                    processId = processId,
                    appDomainId = appDomainId,
                    deliveryAttemptCount = deliveryAttemptCount,
                    errorMsg = exception.ToString()

                });
            }
            catch(Exception exc)
            {
                this.Logger.Log(MethodBase.GetCurrentMethod(), exc);
                throw;
            }
        }

        private string ToDelimitedString<T>(IEnumerable<T> items, char seperator = ',')
        {
            StringBuilder delimitedString = new StringBuilder();
            foreach(var item in items)
            {
                delimitedString.Append(item.ToString());
                delimitedString.Append(seperator);
            }
            return delimitedString.ToString();
        }

        private IEnumerable<string> FromDelimitedString(string delimitedString, char seperator = ',')
        {
            string[] fragments = delimitedString.Split(seperator);
            foreach(var fragment in fragments)
                yield return fragment;
        }
    }
}